package com.service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;

import com.nined.nds.api.DateTimeUtil;
import com.nined.nds.api.JsonUtil;
import com.nined.nds.model.Order;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class SocketServer implements ServerListener {
	protected final Logger log = Logger.getLogger(getClass());
	private int port = 9200;
	private ServerBootstrap serverBootstrap = null;
	private EventLoopGroup bossGroup = null;
	private EventLoopGroup workerGroup = null;
	private Map<Integer, ActiveClient> clients = new HashMap<Integer, ActiveClient>();
    private Map<Integer, NDCallback> callbacks = new HashMap<Integer, NDCallback>(1);
    private Map<Long, SocketMessage> listeners = new HashMap<Long, SocketMessage>();
	private Map<Integer, ActiveClient> lineClients = new HashMap<Integer, ActiveClient>();
	private int sessionId = 1;
	private String sessionKey = String.valueOf(Math.random());
	private boolean started = false;
	public SocketServer(){
		System.out.println("new SocketServer");
	}
	
	protected void finalize(){
		System.out.println("SocketServer finalize");
		close();
	}

	public int getPort() {
		return port;
	}

	public void setPort(int port) {
		this.port = port;
	}
	
	public boolean start(){
		if(started){
			return true;
		}
		try {
			serverBootstrap = new ServerBootstrap();
			bossGroup = new NioEventLoopGroup();
			workerGroup = new NioEventLoopGroup();
			serverBootstrap.group(bossGroup,workerGroup)
				.channel(NioServerSocketChannel.class)
				.option(ChannelOption.SO_BACKLOG, 1024)
				.childHandler(new ChildChannelHandler());
			
			//绑定端口、同步等待
			serverBootstrap.bind(port).sync();
			started = true;
			log.info("socket 已启动.");
			System.out.println("socket started.");
			return true;
		} catch (InterruptedException e) {
			e.printStackTrace();
			started = false;
			return false;
		}catch(Exception ex){
			ex.printStackTrace();
			if(ex.getMessage().contains("Address already in use")){
				started = true;
				return true;
			}
			return false;
		}
	}
	
	public void close(){
		if(started) {
			for(ActiveClient client : clients.values()){
				try{
					SocketMessage resp = new SocketMessage(NDCode.CONNECT_SIGNOUT);
		            send(resp, client.getContext());
				}catch(Exception ex){}
				client.context.close();
			}
			for(ActiveClient client : lineClients.values()) {
				try{
					SocketMessage resp = new SocketMessage(NDCode.CONNECT_SIGNOUT);
		            send(resp, client.getContext());
				}catch(Exception ex){}
				client.context.close();
			}
			bossGroup.shutdownGracefully();
			workerGroup.shutdownGracefully();
			started = false;
			System.out.println("socket close");
		}
	}
	
	private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{
		@Override
		protected void initChannel(SocketChannel ch) throws Exception {
			ch.pipeline().addLast(new ServerHandler(SocketServer.this));
		}
	}
	
	private static class Nested {
		private static final SocketServer instance = new SocketServer();
	}

	public static SocketServer getServer() {
		return Nested.instance;
	}

	synchronized int newSessionId(){
		return sessionId++;
	}
	
	synchronized void addClient(ActiveClient client){
		clients.put(client.getSessionId(), client);
	}
	
	void send(SocketMessage msg, ChannelHandlerContext ctx){
		ctx.writeAndFlush(msg.toByteBuf());
	}
	@Override
	public void onReceived(SocketMessage msg, ChannelHandlerContext ctx) {
		if(msg.getCommandId() > 0 && callbacks.containsKey(msg.getCommandId())){
			if(callbacks.get(msg.getCommandId()).getSuccCode() == msg.getCode() || callbacks.get(msg.getCommandId()).getSuccCode() == NDCode.UNKNOWN){
				callbacks.get(msg.getCommandId()).getCallback().onSucc(msg);//成功
			}else{
				callbacks.get(msg.getCommandId()).getCallback().onFail(msg);//失敗
			}
			callbacks.remove(msg.getCommandId());//移除回调函數
		}
		switch(msg.getCode()){
		case NDCode.CONNECT_SIGNIN:{
			ActiveClient client = new ActiveClient();
			int sessionId = newSessionId();
			client.setSessionId(sessionId);
			client.setSessionKey(sessionKey);
			client.setContext(ctx);
			addClient(client);
			SocketMessage resp = new SocketMessage(NDCode.CONNECT_SIGNIN_OK, msg);
			resp.setSessionId(sessionId);
			resp.setSessionKey(sessionKey);
            send(resp, ctx);
            log.info("客户端连接:" + sessionId);
            System.out.println("client connected.");
			break;
		}
		case NDCode.CONNECT_ACTIVE:{
			SocketMessage resp = new SocketMessage(NDCode.CONNECT_ACTIVE_OK, msg);
            send(resp, ctx);
			break;
		}
		case NDCode.ORDER_PAY_LISTENER:{
			Order order = msg.getBodyObj(Order.class);
			msg.setCreateTime(DateTimeUtil.getNowTimestamp());
			listeners.put(order.getId(), msg);
			log.info("监听订单:" + order.getId());
			System.out.println("listening order " + order.getId() + "...");
			cleanListeners();
			break;
		}
		case NDCode.LINE_PLAY_LISTENER:{
			addLineClients(ctx, msg);
			break;
		}
		}
	}

	public void bizCallback(Long bizId, Object result){
		if(listeners.containsKey(bizId)){
			SocketMessage req = listeners.get(bizId);
			if(clients.containsKey(req.getSessionId())){
				ActiveClient client = clients.get(req.getSessionId());
				SocketMessage msg = new SocketMessage(NDCode.ORDER_PAY_CALLBACK, req);
				if(result != null){
					msg.setBodyJson(JsonUtil.toJson(result));
				}
				send(msg, client.getContext());
				listeners.remove(bizId);
				log.info("回调订单:" + bizId);
				System.out.println("order callback" + bizId);
			}
		}
	}
	
	synchronized void cleanListeners(){
		List<Long> keys = new ArrayList<Long>();
		for(Long key : listeners.keySet()) {
			Integer min = DateTimeUtil.MiniCompare(DateTimeUtil.getNowTimestamp(), listeners.get(key).getCreateTime());
			if(min > 2) {
				keys.add(key);
			}
		}
		for(Long key : keys) {
			listeners.remove(key);
		}
	}
	@SuppressWarnings("rawtypes")
	public void lineCallback(Integer loginId, Object result){
		if(lineClients != null && lineClients.size() > 0) {
			Iterator it = lineClients.entrySet().iterator();
			while (it.hasNext()) {
				Map.Entry entry = (Map.Entry) it.next();
				ActiveClient client = (ActiveClient) entry.getValue();
				if(client.getLoginId() != null && client.getLoginId().equals(loginId)) {
					SocketMessage req = new SocketMessage();
					req.setSessionId(client.getSessionId());
					req.setSessionKey(client.getSessionKey());
				    SocketMessage msg = new SocketMessage(NDCode.LINE_PLAY_CALLBACK, req);
					if(result != null){
						msg.setBodyJson(JsonUtil.toJson(result));
					}
					send(msg, client.getContext());
					log.info("回调在线数据:" + client.getSessionId());
					System.out.println("line callback:" + client.getSessionId());
				}
			}
		}
	}
	
	@Override
	public void onConnected(ChannelHandlerContext ctx) {
		
	}

	@Override
	public void onDisConnected(ChannelHandlerContext ctx) {
		for(ActiveClient client : clients.values() ){
			if(client.getContext() == ctx){
				clients.remove(client.getSessionId());
			}
		}
		for(ActiveClient client : lineClients.values() ){
			if(client.getContext() == ctx){
				lineClients.remove(client.getSessionId());
			}
		}
	}

	@Override
	public void onTimeout(ChannelHandlerContext ctx) {
		
	}

	@Override
	public void onError(ChannelHandlerContext ctx, Throwable cause) {
		
	}
	
	@SuppressWarnings("rawtypes")
	public void addLineClients(ChannelHandlerContext ctx, SocketMessage msg) {
		boolean flag = true;
		Integer loginId = msg.getCommandId();
		if(lineClients != null && lineClients.size() > 0) {
			Iterator it = lineClients.entrySet().iterator();
			while (it.hasNext()) {
				Map.Entry entry = (Map.Entry) it.next();
				ActiveClient client = (ActiveClient) entry.getValue();
				if(client.getLoginId() != null && client.getLoginId().equals(loginId)) {
					flag = false;
					//已存在，更新
					int sessionId = newSessionId();
					client.setSessionId(sessionId);
					client.setSessionKey(sessionKey);
					client.setLoginId(loginId);
					client.setContext(ctx);
					log.info("大屏客户端连接:" + sessionId);
		            System.out.print("client connected.");
		            break;
				}
			}
		}
		if(flag) {
			ActiveClient client = new ActiveClient();
			int sessionId = newSessionId();
			client.setSessionId(sessionId);
			client.setSessionKey(sessionKey);
			client.setLoginId(loginId);
			client.setContext(ctx);
			lineClients.put(sessionId, client);
			log.info("大屏客户端连接:" + sessionId);
            System.out.print("client connected.");
		}
	}
	
	public static void main(String args[]){
		SocketServer.getServer().start();
	}
}
